@claude review

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

Claude finished @bootjp's task in 3m 45s —— View job

Code Review — PR #791: backup: Redis stream encoder (Phase 0a)

Tasks

• Gather context (PR description, changed files list)
• Read internal/backup/redis_stream.go
• Read internal/backup/redis_stream_test.go
• Read internal/backup/redis_string.go diff
• Read related files for context (stream_helpers, storage codec, existing encoders)
• Review: correctness, error handling, data consistency
• Review: performance & memory
• Review: test coverage
• Post final review

The encoder follows the established hash/list/set pattern cleanly and the fail-closed guards, overflow checks, and TTL routing are all correct. Two data-loss issues need fixing before merge; everything else is minor.

Critical — Data loss

1. Duplicate field names silently dropped (redis_stream.go:333-336)

fieldsMap := make(map[string]string, len(e.fields)/xaddPairWidth)
for i := 0; i+1 < len(e.fields); i += xaddPairWidth {
    fieldsMap[e.fields[i]] = e.fields[i+1]   // last-write-wins for dupe names
}

Redis streams explicitly allow duplicate field names within a single entry (unlike hashes). XADD key * event login event logout produces an entry with two event fields, and pb.RedisStreamEntry.Fields is repeated string — an ordered list — precisely to preserve that ordering. Converting to map[string]string silently drops all but the last value for any repeated name. The backup would produce a JSONL that cannot restore the original entry faithfully.

The hash encoder anticipates this with its array-of-records shape and the comment: "Redis hash field names are binary-safe and JSON object keys are not." The same argument applies here. The design doc's example ({"event":"login","user":"alice"}) uses non-colliding names; a format-version bump to [{"name":…,"value":…}] as the PR itself suggests is the right fix. Alternatively, document that streams with duplicate field names produce corrupt backups — but that's a worse outcome.

Per CLAUDE.md convention: a failing test should accompany the fix. e.g.:

// TestRedisDB_StreamDuplicateFieldNamesPreserved pins that duplicate
// field names within one entry survive the encode/decode round-trip.
val := encodeStreamEntryValue(t, "1-0", []string{"k", "v1", "k", "v2"})

Fix this →

2. Non-UTF-8 field names and values silently corrupted (redis_stream.go:302-305)

type streamEntryJSON struct {
    ID     string            `json:"id"`
    Fields map[string]string `json:"fields"`  // ← string, not binary-safe
}

Go's json.Marshal for string coerces invalid UTF-8 bytes to the Unicode replacement character (U+FFFD). Any stream entry whose field name or value contains non-UTF-8 bytes (e.g., a Protobuf-encoded payload, a UUID with raw bytes, arbitrary sensor data) will have those bytes silently destroyed on backup. The hash encoder solves this identically via marshalRedisBinaryValue, which emits {"base64":"…"} for non-UTF-8 values. The same treatment is needed here.

The PR description argues stream field names are "strings rather than binary-safe bytes" — but the live adapter/redis_storage_codec.go stores and retrieves them as []byte via the same protobuf repeated string fields field that hashes use. There is no ASCII/UTF-8 constraint enforced at the wire level.

Fix this →

Medium — Style consistency

3. flushStreams duplicates the flushWideColumnDir boilerplate (redis_stream.go:249-276)

Every other wide-column encoder (flushHashes, flushLists, flushSets) delegates to flushWideColumnDir. flushStreams hand-rolls the identical pattern: nil-check, ensureDir, sorted key iteration, per-key callback. Given that flushWideColumnDir[T any] accepts any state type, streams slot in directly:

func (r *RedisDB) flushStreams() error {
    return flushWideColumnDir(r, r.streams, "streams", func(dir, uk string, st *redisStreamState) error {
        if r.warn != nil && st.metaSeen && int64(len(st.entries)) != st.length {
            r.warn("redis_stream_length_mismatch", ...)
        }
        return r.writeStreamJSONL(dir, []byte(uk), st)
    })
}

Not a bug, but the divergence means a future change to flushWideColumnDir (error policy, security check, etc.) won't automatically propagate to streams.

Fix this →

Minor

4. In-place sort of st.entries contradicts the clone-at-decode rationale (redis_stream.go:323-329)

marshalStreamJSONL calls sort.Slice(st.entries, …), mutating the live state. The self-review point 1 notes that "field slice cloned at decode so snapshot buffer mutations can't bleed into emitted state" — but that same care isn't applied to the entries slice itself. Currently harmless (flush is a one-shot path), but inconsistent. A sort.Slice on a copy — or switching to slices.SortedFunc over a copy — would be strictly consistent.

5. _meta last_ms/last_seq are zero-valued when no meta record arrived

When only entries arrive (test TestRedisDB_StreamEntriesWithoutMetaStillEmitFile), st.metaSeen == false, st.lastMs == 0, st.lastSeq == 0. The emitted _meta line then asserts "last_ms": 0, "last_seq": 0, which a restorer might interpret as "stream never had an entry with ms > 0". This is arguably correct (unknown = zero), but a comment in marshalStreamJSONL explaining the invariant would help a reader of the output format.

6. Large uint64 values lose precision in standard JSON readers (informational)

streamMetaJSON.LastMs, LastSeq are uint64, and streamMetaJSON.Length is int64. Values above 2^53 cannot be represented exactly as IEEE 754 doubles. Standard json.Unmarshal into any (as the tests do with map[string]any) will silently round these. A restorer using json.Decoder.UseNumber() avoids this; worth a note in the format doc or an inline comment.

What's good

• All four error sentinels (ErrRedisInvalidStreamMeta, ErrRedisInvalidStreamEntry, ErrRedisInvalidStreamKey) are correctly wrapped and cover every parse boundary.
• (ms, seq) tuple sort is correct — the comment about "10-0" < "2-0" lexicographically is exactly the right justification.
• TTL routing through kindByKey is clean and consistent with hash/list/set.
• The 14 table-driven tests cover the happy path, error paths, boundary conditions, and warn-sink contract well. The assertStreamMetaTerminator helper keeps per-test bodies tight.
• cloneStringSlice in decodeStreamEntryValue is the right defensive move against protobuf buffer reuse.
• Magic-prefix check before protobuf unmarshal is the correct fail-closed order.

@claude review

Claude finished @bootjp's task in 2m 35s —— View job

Code Review — PR #791: backup: Redis stream encoder (Phase 0a)

Tasks

• Gather context (PR description, changed files list)
• Read internal/backup/redis_stream.go
• Read internal/backup/redis_stream_test.go
• Read internal/backup/redis_string.go diff
• Read related context files (redis_set.go, redis_hash.go, marshalRedisBinaryValue)
• Review: P1 correctness issues (TTL ordering, duplicate fields, binary safety)
• Review: performance & memory
• Review: test coverage gaps
• Post final review

The encoder is well-structured and the two P1 issues from the previous review round have both been fixed and tested. One data-loss issue remains open. Everything else is minor/informational.

Previously Fixed — Confirmed

Duplicate field names (prior P1): streamFieldJSON now uses []streamFieldJSON array-of-records rather than map[string]string. TestRedisDB_StreamDuplicateFieldsPreserved pins the fix. ✓

TTL arrives before stream rows (Codex P1): streamState now calls claimPendingTTL on first registration, draining pendingTTL into st.expireAtMs/st.hasTTL. HandleTTL parks unknown-kind TTLs in pendingTTL and streamState drains them. TestRedisDB_StreamTTLArrivesBeforeRows and TestRedisDB_StreamTTLInlinedFromScanIndex cover both arrival orderings. ✓

Critical — Data loss

Non-UTF-8 field names and values silently corrupted (redis_stream.go:316-319)

streamFieldJSON uses plain string for both name and value:

type streamFieldJSON struct {
    Name  string `json:"name"`
    Value string `json:"value"`
}

Go's json.Marshal silently replaces invalid UTF-8 bytes with the Unicode replacement character (U+FFFD). Redis stream field names and values are stored as []byte in pb.RedisStreamEntry.Fields (a repeated string field in protobuf, which is bytes-on-the-wire). Any stream entry whose field name or value contains non-UTF-8 bytes (raw binary payloads, sensor data, legacy encodings) will be silently corrupted in the backup JSONL — those bytes cannot be recovered at restore time.

The hash encoder solves this exactly: hashFieldRecord uses json.RawMessage for both Name and Value, populated by marshalRedisBinaryValue which emits {"base64":"..."} for non-UTF-8 content (redis_hash.go:235-238). The stream encoder needs the same treatment:

// Change streamFieldJSON to mirror hashFieldRecord:
type streamFieldJSON struct {
    Name  json.RawMessage `json:"name"`
    Value json.RawMessage `json:"value"`
}

// In marshalStreamJSONL, replace the field construction loop:
for i := 0; i+1 < len(e.fields); i += xaddPairWidth {
    nameJSON, err := marshalRedisBinaryValue([]byte(e.fields[i]))
    if err != nil {
        return nil, err
    }
    valueJSON, err := marshalRedisBinaryValue([]byte(e.fields[i+1]))
    if err != nil {
        return nil, err
    }
    fields = append(fields, streamFieldJSON{Name: nameJSON, Value: valueJSON})
}

Per CLAUDE.md convention, a failing test should come first — e.g.:

// TestRedisDB_StreamNonUTF8FieldsRoundTrip pins that a stream entry
// whose field name or value contains non-UTF-8 bytes survives the
// encode step without silent U+FFFD replacement.
val := encodeStreamEntryValue(t, "1-0", []string{"\xff\xfe", "\x00\x01\x02"})

The readStreamJSONL → extractStreamFieldsAsPairs chain would need to handle the {"base64":"..."} shape, but locking the format in a test is the right investment before the decoder is written.

Fix this →

Medium — Style consistency

flushStreams duplicates the flushWideColumnDir boilerplate (redis_stream.go:260-287)

flushSets, flushLists, and flushHashes all delegate to the generic flushWideColumnDir[T]. flushStreams hand-rolls the same pattern: nil-check, ensureDir, sorted user-key slice, per-key callback. A future change to flushWideColumnDir (error policy, security check) won't propagate here automatically. *redisStreamState slots directly into the generic:

func (r *RedisDB) flushStreams() error {
    return flushWideColumnDir(r, r.streams, "streams", func(dir, uk string, st *redisStreamState) error {
        if r.warn != nil && st.metaSeen && int64(len(st.entries)) != st.length {
            r.warn("redis_stream_length_mismatch",
                "user_key_len", len(uk),
                "declared_len", st.length,
                "observed_entries", len(st.entries),
                "hint", "meta record's Length does not match the count of !stream|entry| keys for this user key")
        }
        return r.writeStreamJSONL(dir, []byte(uk), st)
    })
}

Fix this →

Minor

In-place sort of st.entries mutates live state (redis_stream.go:347)

sort.Slice(st.entries, ...) mutates the accumulated slice in-place. Currently harmless (flush is one-shot), but inconsistent with the cloneStringSlice rationale in decodeStreamEntryValue ("field slice cloned so snapshot buffer mutations can't bleed into emitted state"). Sort a shallow copy for consistency.

Large uint64 values lose precision in JSON readers (informational)

streamMetaJSON.LastMs and LastSeq are uint64 encoded as JSON numbers. Values above 2^53 can't be represented as IEEE 754 doubles; a restorer using json.Unmarshal into any (as the test helpers do) would silently round them. For timestamps this won't matter in practice (~year 287,000). Worth a one-line comment on streamMetaJSON noting that restorers should use json.Decoder.UseNumber() for strict fidelity. Note also: the assertStreamMetaTerminator test helper compares via float64 — assertions on seq values above 2^53 would silently pass even if rounded.

What's good

• The pendingTTL / claimPendingTTL infrastructure is a clean, general solution that also retroactively fixes set TTL ordering (backup: Redis set encoder (Phase 0a) #758).
• The (ms, seq) tuple sort is correct with the right justification.
• cloneStringSlice defensive copy in decodeStreamEntryValue is appropriate.
• Magic-prefix check before protobuf unmarshal is fail-closed in the right order.
• All four error sentinels are correctly wrapped.
• 14 table-driven tests cover the happy path, all four error types, TTL inlining (both orderings), empty stream, length-mismatch warning, and sort determinism.
• TestRedisDB_StreamDuplicateFieldsPreserved and TestRedisDB_StreamTTLArrivesBeforeRows are the right regression pins for the two fixed P1s.

@claude review

Claude finished @bootjp's task in 3m 6s —— View job

Code Review — PR #791: backup: Redis stream encoder (Phase 0a) — Round 3

Tasks

• Read trigger comment and prior review context
• Read current internal/backup/redis_stream.go
• Read current internal/backup/redis_stream_test.go
• Read internal/backup/redis_string.go (modified)
• Read internal/backup/redis_set.go (modified)
• Check: non-UTF-8 field corruption fix status
• Check: flushStreams boilerplate dedup status
• Review: any new issues
• Post final review

All three P1 issues from prior rounds are now fixed and regression-tested. One medium issue from both previous reviews is still open. Everything else is minor.

Previously Critical — All Confirmed Fixed

Duplicate field names — streamFieldJSON now uses []streamFieldJSON (array-of-records with json.RawMessage). TestRedisDB_StreamDuplicateFieldsPreserved pins the round-trip. ✓

Non-UTF-8 field corruption — streamFieldJSON.Name/Value are json.RawMessage populated by marshalRedisBinaryValue in buildStreamFieldRecords (redis_stream.go:418-432). TestBuildStreamFieldRecords_NonUTF8BytesRoundTrip pins the projection directly. ✓

TTL arrives before stream rows — streamState calls claimPendingTTL on first registration; TestRedisDB_StreamTTLArrivesBeforeRows and TestRedisDB_StreamTTLInlinedFromScanIndex cover both orderings. setState in redis_set.go:131-134 retroactively applies the same fix to sets. ✓

Medium — Still Open

flushStreams duplicates the flushWideColumnDir boilerplate (redis_stream.go:260-287)

This has been called out in both previous rounds and remains unfixed. flushSets, flushLists, and flushHashes all delegate to the generic flushWideColumnDir[T]. flushStreams still hand-rolls the identical nil-check / ensureDir / sorted-key-slice / per-key-callback pattern. *redisStreamState slots in directly:

func (r *RedisDB) flushStreams() error {
    return flushWideColumnDir(r, r.streams, "streams", func(dir, uk string, st *redisStreamState) error {
        if r.warn != nil && st.metaSeen && int64(len(st.entries)) != st.length {
            r.warn("redis_stream_length_mismatch",
                "user_key_len", len(uk),
                "declared_len", st.length,
                "observed_entries", len(st.entries),
                "hint", "meta record's Length does not match the count of !stream|entry| keys for this user key")
        }
        return r.writeStreamJSONL(dir, []byte(uk), st)
    })
}

A future change to flushWideColumnDir (error policy, security check) won't propagate to the hand-rolled copy.

Fix this →

Minor

Stale file-header comment still shows old map-based field format (redis_stream.go:41-43)

The package doc example shows "fields":{"event":"login","user":"alice"} — the map shape from the design doc that was intentionally abandoned in favour of the [{"name":...,"value":...}] array. A reader scanning the file header would get a misleading picture of the actual output format.

// {"id":"1714400000000-0","fields":{"event":"login","user":"alice"}}

Should be updated to reflect the current array shape, e.g.:

// {"id":"1714400000000-0","fields":[{"name":"event","value":"login"},{"name":"user","value":"alice"}]}

In-place sort of st.entries mutates live state (redis_stream.go:359)

sort.Slice(st.entries, ...) still sorts the accumulated slice in-place. Currently harmless (flush is one-shot), but inconsistent with the cloneStringSlice rationale in decodeStreamEntryValue.

buildStreamFieldRecords binary-safety is a future-only guard (informational)

The test comment at redis_stream_test.go:292-295 correctly identifies a subtlety worth noting in the production code too: proto3 string fields enforce UTF-8 — gproto.Marshal rejects non-UTF-8 bytes — so the {"base64":"..."} branch in the JSONL output cannot be triggered via the live store path today. The marshalRedisBinaryValue guard is the right defensive move (it costs nothing and protects against a future Fields bytes schema migration), but a one-line comment on buildStreamFieldRecords clarifying this would help a reader understand why the test goes directly to buildStreamFieldRecords rather than through HandleStreamEntry.

Large uint64 JSON precision (informational, carried from prior rounds)

streamMetaJSON.LastMs / LastSeq are uint64; assertStreamMetaTerminator compares them as float64. Values above 2^53 silently round in both json.Unmarshal and the test assertions. Not a practical risk for current timestamps, but worth a one-line comment on streamMetaJSON noting that restorers should use json.Decoder.UseNumber() for strict fidelity.

What's good

• All three prior P1s fixed cleanly with companion regression tests per CLAUDE.md convention.
• buildStreamFieldRecords + marshalRedisBinaryValue is the correct binary-safe projection, exactly mirroring hashFieldRecord.
• pendingTTL / claimPendingTTL infrastructure is a general solution that also retroactively corrects the set TTL ordering bug from backup: Redis set encoder (Phase 0a) #758.
• TestBuildStreamFieldRecords_NonUTF8BytesRoundTrip correctly tests the buildStreamFieldRecords function directly (since gproto.Marshal would reject the bytes) and uses extractStreamFieldsAsPairs + decodeRedisBinaryEnvelope to verify the full round-trip. The helpers are clean and reusable.
• TestRedisDB_StreamDuplicateFieldsPreserved, TestRedisDB_StreamTTLArrivesBeforeRows, and TestRedisDB_SetTTLArrivesBeforeRows are the correct regression pins.
• (ms, seq) tuple sort is correct with the right justification; cloneStringSlice defensive copy is appropriate.
• All four error sentinels are correctly wrapped and cover every parse boundary.